package com.huilian.iotbox.mqtt.server.emqx.subscribe;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.huilian.iotbox.aliyun.util.IotUtils;
import com.huilian.iotbox.data.constants.IotConstant;
import com.huilian.iotbox.data.enums.DeviceServiceProviderEnum;
import com.huilian.iotbox.data.enums.IotBoxCmdEnum;
import com.huilian.iotbox.data.enums.MessageLogResultEnum;
import com.huilian.iotbox.data.enums.MessageLogTypeEnum;
import com.huilian.iotbox.data.po.MessageLog;
import com.huilian.iotbox.data.producer.MessageLogRabbit;
import com.huilian.iotbox.data.service.ErrorLogCommonService;
import com.huilian.iotbox.data.utils.IotBoxUtil;
import com.huilian.iotbox.mqtt.server.base.handler.MessageHandlerStrategy;
import com.huilian.iotbox.mqtt.server.base.service.RabbitSenderService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.ObjectUtils;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.PreDestroy;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class AmqpClient implements ApplicationRunner {

    @Autowired
    private ErrorLogCommonService errorLogCommonService;
    @Autowired
    private RabbitSenderService rabbitSenderService;
    @Autowired
    private Map<String, MessageHandlerStrategy> messageHandlerStrategyMap;
    public static List<MqttClient> mqttClientList = new ArrayList<>();

    private final static String STATUS_TOPIC_SUFFIX = "$SYS/brokers/";
    private final static String CONNECTED_STATUS_TOPIC_PREFIX = "/connected";
    private final static String DISCONNECTED_STATUS_TOPIC_PREFIX = "/disconnected";
    private final static String SUBSCRIBED_STATUS_TOPIC_PREFIX = "/subscribed";
    private final static String UNSUBSCRIBED_STATUS_TOPIC_PREFIX = "/unsubscribed";

    //控制台服务端订阅中消费组状态页客户端ID一栏将显示clientId参数。
    //建议使用机器UUID、MAC地址、IP等唯一标识等作为clientId。便于您区分识别不同的客户端。
    private static String clientId = "subscribe";

    @Value("${huilian.iot.productKey}")
    private String productKey;

    @Value("${huilian.iot.host}")
    private String broker;
    // 订阅主题，用的是共享订阅，前缀【$share/huilian/】
    private String subscribeBaseTopic = "$share/huilian//%s/+/user/update";
    private String connectedBaseTopic = "$share/huilian/$SYS/brokers/+/clients/+/connected";
    private String disconnectedBaseTopic = "$share/huilian/$SYS/brokers/+/clients/+/disconnected";
    private String subscribedBaseTopic = "$share/huilian/$SYS/brokers/+/clients/+/subscribed";
    private String unsubscribedBaseTopic = "$share/huilian/$SYS/brokers/+/clients/+/unsubscribed";

    // 自然源卡投
    private String zryKaTouReadReplyBaseTopic = "$share/huilian//%s/+/properties/read/reply";
    private String zryKaTouWriteReplyBaseTopic = "$share/huilian//%s/+/properties/write/reply";
    private String zryKaTouFunctionInvokeBaseTopic = "$share/huilian//%s/+/function/invoke/reply";
    private String zryKaTouEventBaseTopic = "$share/huilian//%s/+/event/+";
    private String zryKaTouReplyBaseTopic = "$share/huilian//%s/+/properties/report";

    private static String[] topicFilters;

    // 订阅账号
    private String username = "huilian-emqx";
    // 订阅密码
    private String password = "YBHGU3zUQRuU1VFjoc5q#8F%S^lUHIT7q!KQNi3xriCkL@m!Rb";
    public static int qos = 0;

    @Autowired
    private MessageLogRabbit messageLogRabbit;
    // 指定单个进程启动的连接数
    // 单个连接消费速率有限，请参考使用限制，最大64个连接
    // 连接数和消费速率及rebalance相关，建议每500QPS增加一个连接
    public static int connectionCount = 8;

    private final static ExecutorService executorService = new ThreadPoolExecutor(
            Runtime.getRuntime().availableProcessors() * 2,
            Runtime.getRuntime().availableProcessors() * 4, 60, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(50000), new CustomizableThreadFactory("emqx-pool-"));

    /**
     * 在这里处理您收到消息后的具体业务逻辑。
     */
    private void processMessage(String topic, MqttMessage mqttMessage) {

        final long beginTime = System.currentTimeMillis();
        MessageLog messageLog = new MessageLog();
        try {
            byte[] body = mqttMessage.getPayload();
            if (body == null) {
                return;
            }
            String content = new String(body);

            final String deviceNo = IotUtils.deviceName(topic, productKey);
            log.info("deviceNo:" + deviceNo + "，content:" + content);
            messageLog.setContent(content);
            messageLog.setMessageId(String.valueOf(mqttMessage.getId()));
            if (StringUtils.isEmpty(deviceNo)) {
                final JSONObject deviceStatus = JSONObject.parseObject(content);
                messageLog.setDeviceNo(deviceStatus.getString("username"));
            } else {
                messageLog.setDeviceNo(deviceNo);
            }
            messageLog.setTopic(topic);
            messageLog.setType(MessageLogTypeEnum.RECEIVE.getCode());
            messageLog.setResult(MessageLogResultEnum.SUCCESS.getCode());
            messageLog.setCreateTime(new Date());
            messageLogRabbit.sendMessageLog(messageLog);

            if (topic.startsWith(STATUS_TOPIC_SUFFIX) && (topic.endsWith(CONNECTED_STATUS_TOPIC_PREFIX) || topic.endsWith(DISCONNECTED_STATUS_TOPIC_PREFIX))) {
                final JSONObject contentJSON = JSONObject.parseObject(content);
                String dNo = contentJSON.getString("username");
                if (username.equals(dNo)) {
                    log.warn("消费者，不执行");
                    return;
                }
                MessageHandlerStrategy messageHandlerStrategy = messageHandlerStrategyMap.get("messageHandleService99");
                if (StringUtils.isEmpty(messageHandlerStrategy)) {
                    log.error(deviceNo + "消息找不到对应的策略: messageHandleService99");
                    return;
                }
                //执行策略
                messageHandlerStrategy.process(content, deviceNo, productKey, DeviceServiceProviderEnum.EMQX.getCode());
            } else if (topic.startsWith(STATUS_TOPIC_SUFFIX) && (topic.endsWith(SUBSCRIBED_STATUS_TOPIC_PREFIX) || topic.endsWith(UNSUBSCRIBED_STATUS_TOPIC_PREFIX))) {
                log.info("subscribed 或 unsubscribed");
            } else if (topic.equalsIgnoreCase(String.format(IotConstant.UPDATE_TOPIC_SUFFIX, productKey, deviceNo))) {
                content = content.toUpperCase();
                if (content.startsWith(IotConstant.CONTENT_PREFIX) && content.endsWith(IotConstant.CONTENT_SUFFIX)) {
                    final String cmd = content.substring(6, 8);
                    final String from = content.substring(4, 6);
                    if (!ObjectUtils.equals("01", from)) {
                        log.error(deviceNo + "消息来源非法: {}", content);
                        errorLogCommonService.writerInfoErrorLog(deviceNo + "消息来源非法", content);
                        return;
                    }
                    final String checkData = content.substring(2, content.length() - 4);
                    final String checkCode = content.substring(content.length() - 4, content.length() - 2);
                    if (!ObjectUtils.equals(checkCode, IotBoxUtil.checkcode_0007(checkData))) {
                        log.error(deviceNo + "消息校验不通过: {}", content);
                        errorLogCommonService.writerInfoErrorLog(deviceNo + "消息校验不通过", content);
                        return;
                    }

                    final MessageHandlerStrategy messageHandlerStrategy = messageHandlerStrategyMap.get("messageHandleService" + cmd);
                    if (cmd.equalsIgnoreCase(IotBoxCmdEnum.AA01DD.getHexadecimal())|| cmd.equalsIgnoreCase(IotBoxCmdEnum.AA33DD.getHexadecimal())) {
                        JSONObject deviceStatus = new JSONObject();
                        deviceStatus.put("cmd", cmd);
                        deviceStatus.put("deviceNo", deviceNo);
                        deviceStatus.put("serviceProviderCode", DeviceServiceProviderEnum.EMQX.getCode());
                        deviceStatus.put("content", content);
                        rabbitSenderService.deviceHeartbeat(deviceStatus);
                        return;
                    }

                    if (StringUtils.isEmpty(messageHandlerStrategy)) {
                        log.error(deviceNo + "消息找不到对应的策略: {}", content);
                        errorLogCommonService.writerInfoErrorLog(deviceNo + "消息找不到对应的策略", content);
                        return;
                    }
                    log.info(deviceNo + "接收到消息: {}", content);
                }
            } else {
                log.error(deviceNo + "接收到消息无法命中: {}", content);
            }
        } catch (Exception e) {
            e.printStackTrace();
            String deviceNo = !StringUtils.isEmpty(messageLog.getDeviceNo()) ? messageLog.getDeviceNo() : "设备号未知";
            String content = !StringUtils.isEmpty(messageLog.getContent()) ? messageLog.getContent() : "消息内容未知";

            errorLogCommonService.writerInfoException(String.format("信息接收失败，%s，%s", deviceNo, content), messageLog, e);
            log.error("信息接收失败：{}", e.getMessage());
        } finally {
            final Long totalTime = System.currentTimeMillis() - beginTime;
            if (totalTime.compareTo(100L) >= 0) {
                log.info("业务总流程消耗时间：" + (totalTime) + "ms");
            }
        }
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {

        if (connectionCount <= 1) {
            throw new RuntimeException("connectionCount个数 >= 2");
        }

        final String subscribeTopic = String.format(subscribeBaseTopic, productKey);
        log.info("subscribeTopic:" + subscribeTopic);

        topicFilters = new String[]{
                subscribeTopic,
                connectedBaseTopic,
                disconnectedBaseTopic,
                subscribedBaseTopic,
                unsubscribedBaseTopic,
                String.format(zryKaTouReadReplyBaseTopic, productKey),
                String.format(zryKaTouWriteReplyBaseTopic, productKey),
                String.format(zryKaTouFunctionInvokeBaseTopic, productKey),
                String.format(zryKaTouEventBaseTopic, productKey),
                String.format(zryKaTouReplyBaseTopic, productKey)
        };

        for (int i = 0; i < connectionCount; i++) {
            StringBuffer clientIdSB = new StringBuffer(clientId)
                    .append("-")
                    .append(UUID.randomUUID());
            MqttClient client = new MqttClient(broker, clientIdSB.toString(), new MemoryPersistence());
            // 设置连接参数
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true);
            options.setUserName(username);
            options.setPassword(password.toCharArray());
            // 设置超时时间 单位为秒
            options.setConnectionTimeout(1000);
            // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
            options.setKeepAliveInterval(240);
            // 设置回调
            client.setCallback(new MqttCallback() {

                public void connectionLost(Throwable cause) {
                    log.error(client.getClientId() + ": " + cause.getMessage());
                }

                public void messageArrived(String topic, MqttMessage mqttMessage) {
                    executorService.submit(() -> processMessage(topic, mqttMessage));
                }

                public void deliveryComplete(IMqttDeliveryToken token) {
                    if (!token.isComplete()) {
                        log.error(client.getClientId() + "投递失败: " + JSON.toJSONString(token));
                        errorLogCommonService.writerInfoErrorLog(client.getClientId() + "投递失败: ", JSON.toJSONString(token));
                    }
                }

            });
            client.connect(options);
            client.subscribe(topicFilters);
            log.info("EMQX " + clientIdSB + " 连接成功，开始订阅");

            mqttClientList.add(client);
        }


    }

    @PreDestroy
    public void shutDownPool() throws InterruptedException {
        for (MqttClient mqttClient : mqttClientList) {
            try {
                log.info(mqttClient.getClientId() + "关闭订阅端");
                mqttClient.disconnect();
            } catch (Exception e) {
                log.error(mqttClient.getClientId() + ":failed to close connection:", e);
                e.printStackTrace();
            } finally {
                try {
                    mqttClient.close();
                } catch (MqttException e) {
                    e.printStackTrace();
                }
            }
        }

        log.info("关闭线程池");
        executorService.shutdown();
        if (executorService.awaitTermination(10, TimeUnit.SECONDS)) {
            log.info("shutdown success");
        } else {
            log.info("failed to handle messages");
        }
    }
}
